#include "ConsistentHashEventThreadPool.h"

#include <memory>

ConsistentHashRing::ConsistentHashRing(const std::vector<std::string>& nodes, int numReplicas)
{
  m_replicas = numReplicas;
  for (const auto& node : nodes) {
    addNode(node);
  }
  std::sort(m_ring.begin(), m_ring.end());
}

void ConsistentHashRing::addNode(const std::string& node)
{
  for (int i = 0; i < m_replicas; i++) {
    std::string virtualNode = node + "-" + std::to_string(i);
    uint32_t hash = hashFunction(virtualNode);
    m_ring.emplace_back(virtualNode, hash);
  }
}

void ConsistentHashRing::removeNode(const std::string& node) {
  auto it = m_ring.begin();
  while (it != m_ring.end()) {
    if (it->m_key == node) {
      m_ring.erase(it);
      break;
    } else {
      it++;
    }
  }
}

std::string ConsistentHashRing::getNode(const std::string& key) {
  if (m_ring.empty()) {
    return "";
  }
  uint32_t hash = hashFunction(key);
  auto it = std::lower_bound(m_ring.begin(), m_ring.end(), Node(key, hash));
  if (it == m_ring.end()) {
    it = m_ring.begin();
  }
  return it->m_key;
}

ConsistentHashEventLoopThreadPool::ConsistentHashEventLoopThreadPool(EventLoop *base,
                                                                     int numThreads,
                                                                     const std::string& name)
  : EventLoopThreadPool(base, numThreads, name)
{
  setThreadNum(numThreads); 
}

void ConsistentHashEventLoopThreadPool::setThreadNum(int threadNum)
{
  // 先调用父类的 setThreadNum
  EventLoopThreadPool::setThreadNum(threadNum);
  std::vector<std::string> nodeKeys;
  m_nodeMap.clear();
  std::vector<EventLoop*> loops;
  {
    std::unique_lock<std::mutex> ulk(m_mutex);
    loops = getAllLoop();
  }
  for (size_t i = 0; i < loops.size(); i++) {
    std::string nodeKey = "subLoop-" + std::to_string(i);
    nodeKeys.emplace_back(nodeKey);
    m_nodeMap[nodeKey] = loops[i];
  }
  m_hashRing = std::unique_ptr<ConsistentHashRing>(new ConsistentHashRing(nodeKeys));
}

EventLoop* ConsistentHashEventLoopThreadPool::getNextLoop(const std::string& key) {
  EventLoop* loop = m_baseLoop;
  std::string node = m_hashRing->getNode(genHashCode(key));
  auto it = m_nodeMap.find(node);
  if (it!= m_nodeMap.end()) {
    loop = it->second;
  }
  return loop;
}

EventLoop* ConsistentHashEventLoopThreadPool::getNextLoop()
{
  return EventLoopThreadPool::getNextLoop();
}